package com.mxd.flink.connector.table;

import com.mxd.flink.connector.config.RedisOptions;
import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.table.api.TableSchema;
import org.apache.flink.table.catalog.ResolvedSchema;
import org.apache.flink.table.connector.ChangelogMode;
import org.apache.flink.table.connector.format.DecodingFormat;
import org.apache.flink.table.connector.sink.DynamicTableSink;
import org.apache.flink.table.connector.source.*;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.types.DataType;
import org.apache.flink.types.RowKind;

import java.util.Map;

/**
 * @author rongdi
 * @date 2022/9/18 19:50
 */
public class RedisDynamicTableSource implements ScanTableSource, LookupTableSource {

    private ReadableConfig config;

    private ResolvedSchema resolvedSchema;

    private DecodingFormat<DeserializationSchema<RowData>> decodingFormat;

    private DataType producedDataType;

    public RedisDynamicTableSource(ReadableConfig config, ResolvedSchema resolvedSchema, DecodingFormat<DeserializationSchema<RowData>> decodingFormat) {
        this.config = config;
        this.resolvedSchema = resolvedSchema;
        this.decodingFormat = decodingFormat;
        this.producedDataType = resolvedSchema.toPhysicalRowDataType();
    }

    /**
     * 这里就是真正执行遍历数据的时候的函数实现了，一般用在流表的时候会用到，比如redis中一个key的值是一个
     * 数据一直增加的list，每次取数据都从这个list里pop出一条或者多条数据
     * @param scanContext
     * @return
     */
    @Override
    public ScanRuntimeProvider getScanRuntimeProvider(ScanContext scanContext) {
        DeserializationSchema<RowData> deserializer = null;
        if(this.decodingFormat != null) {
            deserializer = this.decodingFormat.createRuntimeDecoder(scanContext, this.producedDataType);
        }
        return SourceFunctionProvider.of(new RedisScanFunction(this.config, deserializer), false);
    }

    /**
     * 这里就是真正执行点查询的函数实现了，一般在flinksql中作为维表的时候使用点查询
     * 执行点查询的LookupRuntimeProvider必须是TableFunction（同步）或者AsyncTableFunction（异步）
     * @param lookupContext
     * @return
     */
    @Override
    public LookupRuntimeProvider getLookupRuntimeProvider(LookupContext lookupContext) {
        DeserializationSchema<RowData> deserializer = null;
        if(this.decodingFormat != null) {
            deserializer = this.decodingFormat.createRuntimeDecoder(lookupContext, this.producedDataType);
        }
        /**
         * 判断是否配置异步开关，如果异步开关打开，则使用异步算子
         */
        boolean asyncEnabled = config.get(RedisOptions.LOOKUP_ASYNC_ENABLED);
        if(asyncEnabled) {
            return AsyncTableFunctionProvider.of(new AsyncRedisLookupFunction(this.config, this.resolvedSchema, deserializer));
        } else {
            return TableFunctionProvider.of(new RedisLookupFunction(this.config, this.resolvedSchema, deserializer));
        }
    }

    /**
     * 这里获取changelog模式，只是一个source，所以直接用一个insert就好了
     * 这里判断decodingFormat是不是为空，是为了兼容参数中没有传format的时候
     * @return
     */
    @Override
    public ChangelogMode getChangelogMode() {
        return this.decodingFormat == null ?
                ChangelogMode.newBuilder().addContainedKind(RowKind.INSERT).build() : this.decodingFormat.getChangelogMode();
    }

    @Override
    public DynamicTableSource copy() {
        return new RedisDynamicTableSource(this.config, this.resolvedSchema,  this.decodingFormat);
    }

    @Override
    public String asSummaryString() {
        return "Redis Dynamic Table Source";
    }
}
